package jmine.tec.batch.event;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import jmine.tec.batch.BatchController;
import jmine.tec.batch.event.domain.Event;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.DisposableBean;

import bancosys.tec.persist.controller.TransactionalController;

/**
 * Fila de eventos a ser processada. Esta fila tem uma capacidade maxima, definida pelo construtor. Eventos inseridos além da capacidade
 * ficam em uma fila de elementos agendados para insercao.
 * <p>
 * 
 * @author takeshi
 */
public class EventQueue implements DisposableBean {

    private static final int WAIT_POLL_TIME = 3000;

    private static final Logger LOGGER = Logger.getLogger(EventQueue.class);

    private final BatchController batchController;

    private final int maxTransactionSize;

    private final AtomicBoolean active = new AtomicBoolean();

    private static final int MAX_ELEMENTS = 1000;

    private static final int TRANSACTION_SIZE = 100;

    private final BlockingQueue<Event> currentQueue;

    private ScheduledExecutorService scheduledExecutor;

    /**
     * @param controller the kernel controller
     */
    public EventQueue(BatchController controller) {
        this(controller, MAX_ELEMENTS, TRANSACTION_SIZE);
    }

    /**
     * @param batchController {@link BatchController}
     * @param maxElements int
     * @param maxTransactionSize int
     */
    public EventQueue(BatchController batchController, int maxElements, int maxTransactionSize) {
        super();
        this.batchController = batchController;
        this.maxTransactionSize = maxTransactionSize;
        this.currentQueue = new LinkedBlockingQueue<Event>(maxElements);
    }

    /**
     * Inicia a thread de eventos se este nao foi inicializado ainda.
     */
    public void start() {
        if (this.active.compareAndSet(false, true)) {
            LOGGER.debug("starting EventConsumerThread");
            this.scheduledExecutor = Executors.newScheduledThreadPool(1, new ThreadFactory() {
                public Thread newThread(Runnable r) {
                    Thread t = new Thread(r, "EventConsumerThread");
                    t.setDaemon(true);
                    t.setPriority(Thread.MIN_PRIORITY);
                    return t;
                }
            });
        }
    }

    /**
     * Faz com que este queue bloqueie novos eventos imediatamente e faz com que a Event Consumer Thread pare de processar eventos
     * imediatamente. Este metodo vai devolver eventos que nao foram processados.
     * 
     * @return {@link List} de {@link Event}
     */
    public List<Event> stopImmediatelly() {
        if (this.active.compareAndSet(true, false)) {
            List<Runnable> list = this.scheduledExecutor.shutdownNow();
            LOGGER.warn("ignoring " + list.size() + " jobs");
            List<Event> queued = new ArrayList<Event>(this.currentQueue.size());
            this.currentQueue.drainTo(queued);
            this.scheduledExecutor = null;
            return queued;
        }
        return new LinkedList<Event>();
    }

    /**
     * Faz com que esta queue bloqueie imediatamente novos eventos, espere até que todos os eventos sejam consumidos.
     * 
     * @throws InterruptedException se a thread atual for interrompida antes da EventConsumerThread terminar de processar todos os eventos
     */
    public void requestStop() throws InterruptedException {
        // forbid new events
        if (this.active.compareAndSet(true, false)) {
            try {
                final boolean debug = LOGGER.isDebugEnabled();
                if (debug) {
                    LOGGER.debug("stopping EventConsumerThread");
                }
                int size = this.currentQueue.size();
                if (debug) {
                    LOGGER.debug("requesting EventConsumerThread to process all and stop");
                    LOGGER.debug("throwing " + size + "messages");
                }
                Future<?> future = this.scheduledExecutor.submit(new Runnable() {
                    public void run() {
                        List<Event> events = new ArrayList<Event>();
                        EventQueue.this.currentQueue.drainTo(events);
                        EventQueue.this.executeAll(events);
                    }
                });
                future.get();
                this.scheduledExecutor.shutdown();
                this.scheduledExecutor.awaitTermination(WAIT_POLL_TIME, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                this.scheduledExecutor.shutdownNow();
                Thread.currentThread().interrupt();
                throw e;
            } catch (ExecutionException e) {
                LOGGER.warn("error waiting for event", e);
            } finally {
                this.scheduledExecutor = null;
            }
        }
    }

    /**
     * Adiciona um evento a fila, devolvendo <code>true</code> se a operacao foi realizada com sucesso e notifica threads esperando na fila,
     * chamando {@link #notifyEventQueued()}
     * 
     * @param evt o evento a ser adicionado
     * @return boolean
     */
    private boolean addToQueue(Event evt) {
        if (this.currentQueue.offer(evt)) {
            if (this.currentQueue.size() >= this.maxTransactionSize) {
                this.notifyEventQueued();
            } else {
                this.scheduledExecutor.schedule(this.consumeTask, WAIT_POLL_TIME, TimeUnit.MILLISECONDS);
            }
            return true;
        }
        return false;
    }

    private final Runnable consumeTask = new Runnable() {
        public void run() {
            List<Event> events = new ArrayList<Event>(EventQueue.this.maxTransactionSize);
            EventQueue.this.currentQueue.drainTo(events, EventQueue.this.maxTransactionSize);
            EventQueue.this.executeAll(events);
            if (!EventQueue.this.currentQueue.isEmpty()) {
                EventQueue.this.scheduledExecutor.submit(EventQueue.this.consumeTask);
            }
        }
    };

    /**
     * Adiciona um evento a esta fila. Se a fila esta cheia, a thread atual "ajuda" a EventConsumerThread, processando eventos na fila e o
     * evento passado como parametro.
     * 
     * @param evt {@link Event}
     * @return boolean
     */
    public boolean throwEvent(final Event evt) {
        if (this.active.get()) {
            if (!this.addToQueue(evt)) {
                LOGGER.debug("offer failed - caller thread will handle events");
                this.scheduledExecutor.submit(this.consumeTask);
                this.scheduledExecutor.schedule(new Runnable() {
                    public void run() {
                        EventQueue.this.throwEvent(evt);
                    }
                }, WAIT_POLL_TIME, TimeUnit.MILLISECONDS);
            }
            return true;
        }
        return false;
    }

    /**
     * Executa todos os eventos da lista passada, em uma transacao. A transacao toma rollback caso alguma excecao seja lancada, porem as
     * excecoes sao ignoradas.
     * 
     * @param list {@link List}
     */
    private void executeAll(List<? extends Event> list) {
        this.getTransactionalController().startTransaction();
        boolean commit = true;
        try {
            for (Event event : list) {
                event.execute(this.batchController);
            }
        } catch (Throwable e) {
            LOGGER.error("Falha ao salvar eventos assíncronos", e);
            commit = false;
        } finally {
            try {
                if (commit) {
                    this.getTransactionalController().commit();
                } else {
                    this.getTransactionalController().rollback();
                }
            } catch (Throwable t) {
                LOGGER.error("Falha ao salvar eventos assíncronos", t);
            }
        }
    }

    /**
     * @return a {@link TransactionalController}
     */
    private TransactionalController getTransactionalController() {
        return this.batchController.getTransactionalController();
    }

    /**
     * callback do {@link DisposableBean} para garantir que, se este bean for gerenciado pelo spring, quando a beanfactory eh destruida,
     * este bean sera destruido tambem.
     */
    public void destroy() {
        this.stopImmediatelly();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String toString() {
        return "EventQueue active: " + this.active.get() + " - " + this.currentQueue;
    }

    /**
     * 
     */
    public void notifyEventQueued() {
        this.scheduledExecutor.execute(this.consumeTask);
    }

}
